package soar.basic;

import java.util.Iterator;
import java.util.ArrayList;
import java.util.LinkedList;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SelectionKey;

public class Ep
{
    private static final String __CLASS__=Ep.class.getName();

    public enum SlotType { CONN, RECV, ABORT, TIMEOUT }

    public static final int POLL_MAX_NUM=1024;

    private static final int READ_BUFFER_SIZE=128*1024;

    private Selector selector;

    private class ReadCache
    {/*{{{*/
        private static final int READ_CACHE_DEF_SIZE=1024; 

        private byte[] msg;
        private int offset;
        private int end;

        ReadCache()
        {
            this.msg=new byte[READ_CACHE_DEF_SIZE];
            this.offset=0;
            this.end=0;
        }

        void append(byte[] newmsg, int size)
        {
            if(this.msg.length-this.end >= size)
            {
                System.arraycopy(newmsg, 0, this.msg, this.end, size);
                this.end+=size;
            }
            else
            {
                int remain_size=this.end-this.offset;
                byte[] remain=this.msg;
                this.msg=new byte[remain_size+size];
                System.arraycopy(remain, 0, this.msg, 0, remain_size);
                System.arraycopy(newmsg, 0, this.msg, remain_size, size);
                this.offset=0;
                this.end=remain_size+size;
            }
        }

        int size()
        {
            return this.end-this.offset;
        }

        int getMsgSize()
        {
            byte[] buf=new byte[Sock.MSG_SIZE_SIZE];
            System.arraycopy(this.msg, this.offset, buf, 0,
                    Sock.MSG_SIZE_SIZE);
            String s=new String(buf);
            //Errlog.add(__CLASS__+": msg_size(%s)", s);
            int msg_size;
            try
            {
                msg_size=Integer.parseInt(s, 16);
            }
            catch(Exception e)
            {
                return -1;
            }
            if(msg_size > Sock.MSG_MAX_SIZE)
            {
                return -1;
            }
            return msg_size;
        }

        byte[] getMsg(int msg_size)
        {
            byte[] msg=new byte[msg_size];
            System.arraycopy(this.msg, this.offset+Sock.MSG_SIZE_SIZE,
                    msg, 0, msg_size);
            this.offset+=Sock.MSG_SIZE_SIZE+msg_size;
            //Errlog.add(__CLASS__+": offset(%d) end(%d)",
            //        this.offset, this.end);
            if(this.offset == this.end)
            {
                this.offset=this.end=0;
            }
            return msg;
        }
    }/*}}}*/

    private class WriteNode
    {/*{{{*/
        public byte[] msg;
        public int offset;
    }/*}}}*/

    private class Slot
    {/*{{{*/
        int slot_id;

        ServerSocketChannel server_channel;

        boolean is_wrap;
        SocketChannel channel;
        int timeout;
        long last_ts;
        ReadCache read_cache;
        LinkedList<WriteNode> write_list;

        Slot(int slot_id)
        {
            this.slot_id=slot_id;
            this.server_channel=null;
            this.is_wrap=false;
            this.channel=null;
            this.timeout=0;
            this.last_ts=System.currentTimeMillis();
            this.read_cache=null;
            this.write_list=null;
        }
    }/*}}}*/
    private Slot[] slots;
    private int slot_upper_id;

    private ByteBuffer read_buffer;

    public Ep()
    {/*{{{*/
        try
        {
            this.selector=Selector.open();
        }
        catch(Exception e)
        {
        }
        this.slots=new Slot[POLL_MAX_NUM];
        this.slot_upper_id=0;
        this.read_buffer=ByteBuffer.allocate(READ_BUFFER_SIZE);
    }/*}}}*/

    public int addListen(ServerSocketChannel channel)
    {/*{{{*/
        if(channel == null)
        {
            return -1;
        }

        int slot_id;
        for(slot_id=0; slot_id<this.slot_upper_id; ++slot_id)
        {
            if(this.slots[slot_id] == null) break;
        }
        if(slot_id == this.slot_upper_id)
        {
            if(this.slot_upper_id == this.slots.length)
                return -2;
            ++this.slot_upper_id;
        }
        this.slots[slot_id]=new Slot(slot_id);
        this.slots[slot_id].server_channel=channel;

        try
        {
            SelectionKey key=channel.register(this.selector,
                    SelectionKey.OP_ACCEPT);
            key.attach(new Integer(slot_id));
        }
        catch(Exception e)
        {
            this.slots[slot_id]=null;
            return -3;
        }

        return slot_id;
    }/*}}}*/

    public int add(SocketChannel channel)
    {/*{{{*/
        if(channel == null)
        {
            return -1;
        }

        int slot_id;
        for(slot_id=0; slot_id<this.slot_upper_id; ++slot_id)
        {
            if(this.slots[slot_id] == null) break;
        }
        if(slot_id == this.slot_upper_id)
        {
            if(this.slot_upper_id == this.slots.length)
                return -2;
            ++this.slot_upper_id;
        }
        this.slots[slot_id]=new Slot(slot_id);
        this.slots[slot_id].channel=channel;
        this.slots[slot_id].read_cache=new ReadCache();
        this.slots[slot_id].write_list=new LinkedList<WriteNode>();

        try
        {
            channel.configureBlocking(false);
            SelectionKey key=channel.register(this.selector,
                    SelectionKey.OP_READ);
            key.attach(new Integer(slot_id));
        }
        catch(Exception e)
        {
            this.slots[slot_id]=null;
            return -3;
        }

        return slot_id;
    }/*}}}*/

    public boolean setWrap(int slot_id)
    {/*{{{*/
        if(slot_id < 0 || slot_id > POLL_MAX_NUM)
            return false;
        if(this.slots[slot_id] == null)
            return false;
        if(this.slots[slot_id].channel == null)
            return false;
        this.slots[slot_id].is_wrap=true;
        return true;
    }/*}}}*/

    public boolean setTimeout(int slot_id, int timeout_sec)
    {/*{{{*/
        if(slot_id < 0 || slot_id > POLL_MAX_NUM)
            return false;
        if(this.slots[slot_id] == null)
            return false;
        if(this.slots[slot_id].channel == null)
            return false;
        this.slots[slot_id].timeout=timeout_sec*1000;
        return true;
    }/*}}}*/

    public class ActiveSlot
    {/*{{{*/
        public SlotType type;
        public int slot_id;

        public class Conn
        {
            public SocketChannel channel;  // CONN
            public String addr;
        }
        public Conn conn;

        public class Recv
        {
            public byte[] msg;
        }
        public Recv recv;

        ActiveSlot(SlotType type, int slot_id,
                SocketChannel new_channel)
        {
            this.type=type;
            this.slot_id=slot_id;
            this.conn=new Conn();
            this.conn.channel=new_channel;
            Socket sock=new_channel.socket();
            this.conn.addr=sock.getInetAddress().getHostAddress()+":"+
                sock.getPort();
        }

        ActiveSlot(SlotType type, int slot_id, byte[] msg)
        {
            this.type=type;
            this.slot_id=slot_id;
            this.recv=new Recv();
            this.recv.msg=msg;
        }

        ActiveSlot(SlotType type, int slot_id)
        {
            this.type=type;
            this.slot_id=slot_id;
        }
    }/*}}}*/

    private void fill(int slot_id, SelectionKey key,
            ArrayList<ActiveSlot> active)
    {/*{{{*/
        Slot slot=this.slots[slot_id];

        int count;
        try
        {
            count=slot.channel.read(this.read_buffer);
            //Errlog.add(__CLASS__+": read %d", count);
            if(count < 0)
            {
                //Errlog.add(__CLASS__+": slot(%d) read %d, revoke",
                //        slot_id, count);
                active.add(new ActiveSlot(SlotType.ABORT, slot_id));
                slot.channel.close();
                this.slots[slot_id]=null;
                return;
            }
        }
        catch(Exception e)
        {
            Errlog.add(__CLASS__+": slot(%d) read fail: %s",
                    slot_id, e.toString());
            return;
        }

        this.read_buffer.flip();
        byte[] msg=this.read_buffer.array();
        this.read_buffer.clear();
        if(slot.is_wrap)
        {
            ReadCache rc=slot.read_cache;
            rc.append(msg, count);
            while(rc.size() > 0)
            {
                if(rc.size() <= Sock.MSG_SIZE_SIZE)
                    break;
                int msg_size=rc.getMsgSize(); 
                //Errlog.add(__CLASS__+": msg_size(%d)", msg_size);
                if(msg_size < 0)
                {
                    //Errlog.add(__CLASS__+": get msg_size(%d), close",
                    //        msg_size);
                    active.add(new ActiveSlot(SlotType.ABORT, slot_id));
                    try
                    {
                        slot.channel.close();
                    }
                    catch(Exception e)
                    {
                    }
                    this.slots[slot_id]=null;
                    return;
                }
                if(Sock.MSG_SIZE_SIZE+msg_size > rc.size())
                {
                    //Errlog.add(__CLASS__+": MSG_SIZE_SIZE+msg_size(%d)>"+
                    //        "remain_size(%d)",
                    //        msg_size, rc.size());
                    break;
                }
                active.add(new ActiveSlot(SlotType.RECV, slot_id,
                            rc.getMsg(msg_size)));
            }
        }
        else
        {
            active.add(new ActiveSlot(SlotType.RECV, slot_id, msg));
        }
    }/*}}}*/

    private void flush(int slot_id, SelectionKey key)
    {/*{{{*/
        if(this.slots[slot_id] == null)
        {
            return;
        }

        Slot slot=this.slots[slot_id];
        if(slot.write_list.isEmpty())
        {
            return;
        }
        while(!slot.write_list.isEmpty())
        {
            WriteNode node=slot.write_list.poll();
            int offset=0;
            try
            {
                int length=node.msg.length-node.offset;
                ByteBuffer bb=ByteBuffer.wrap(node.msg, node.offset,
                        length);
                int count=slot.channel.write(bb);
                //Errlog.add(__CLASS__+": flush send(%d)", count);
                if(count < length)
                {
                    node.offset+=count;
                    slot.write_list.offerFirst(node);
                    return;
                }
            }
            catch(Exception e)
            {
                Errlog.add(__CLASS__+": slot(%d) flush fail: %s",
                        slot_id, e.toString());
                return;
            }
        }
        key.interestOps(key.interestOps() & ~SelectionKey.OP_WRITE);
    }/*}}}*/

    public ArrayList<ActiveSlot> poll()
    {/*{{{*/
        ArrayList<ActiveSlot> active=new ArrayList<ActiveSlot>();

        int slot_id;
        long now=System.currentTimeMillis();
        for(slot_id=0; slot_id<this.slot_upper_id; ++slot_id)
        {
            Slot slot=this.slots[slot_id];
            if(slot != null && slot.channel != null &&
                    slot.timeout > 0 &&
                    now - slot.last_ts > slot.timeout)
            {
                //Errlog.add(__CLASS__+": slot(%d) timeout", slot_id);
                active.add(new ActiveSlot(SlotType.TIMEOUT, slot_id));
                try
                {
                    slot.channel.close();
                }
                catch(Exception e)
                {
                }
                this.slots[slot_id]=null;
            }
        }

        try
        {
            this.selector.select(1000);
            Iterator it=this.selector.selectedKeys().iterator();
            while(it.hasNext())
            {
                SelectionKey key=(SelectionKey)it.next();
                it.remove();
                Integer id=(Integer)key.attachment();
                slot_id=id.intValue();
                //Errlog.add("select slot_id(%d)", slot_id);

                if(key.isAcceptable())
                {
                    ServerSocketChannel server_channel=
                        this.slots[slot_id].server_channel;
                    SocketChannel new_channel=server_channel.accept();
                    if(new_channel != null)
                    {
                        active.add(new ActiveSlot(SlotType.CONN, slot_id,
                                    new_channel)); 
                    }
                }
                else
                {
                    if(this.slots[slot_id].timeout > 0)
                    {
                        this.slots[slot_id].last_ts=
                            System.currentTimeMillis();
                    }
                    if(key.isReadable())
                    {
                        fill(slot_id, key, active);
                    }
                    if(key.isValid() && key.isWritable())
                    {
                        Errlog.add(__CLASS__+": slot(%d) writable",
                                slot_id);
                        flush(slot_id, key);
                    }
                }
            }
        }
        catch(Exception e)
        {
            Errlog.add(__CLASS__+": select fail: %s", e.toString());
        }

        return active;
    }/*}}}*/

    public boolean send(int slot_id, byte[] msg)
    {/*{{{*/
        if(slot_id < 0 || slot_id > POLL_MAX_NUM)
            return false;
        Slot slot=this.slots[slot_id];
        if(slot == null)
        {
            return false;
        }
        if(slot.channel == null)
        {
            return false;
        }
        if(slot.is_wrap)
            msg=Sock.wrap(msg);
        if(slot.timeout > 0)
            slot.last_ts=System.currentTimeMillis();

        int offset=0;
        if(slot.write_list.isEmpty())
        {
            try
            {
                ByteBuffer bb=ByteBuffer.wrap(msg);
                offset=slot.channel.write(bb);
                if(offset == msg.length) return true;
            }
            catch(Exception e)
            {
                return false;
            }
        }
        //Errlog.add(__CLASS__+": send msg(%d) sent(%d)",
        //        msg.length, offset);
        WriteNode node=new WriteNode();
        node.msg=msg;
        node.offset=offset;
        slot.write_list.add(node);
        try
        {
            //slot.channel.register(this.selector, SelectionKey.OP_WRITE);
            SelectionKey key=slot.channel.keyFor(this.selector);
            key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
        }
        catch(Exception e)
        {
            Errlog.add(__CLASS__+": slot(%d) register OP_WRITE fail: %s",
                    slot_id, e.toString());
        }

        return true;
    }/*}}}*/
}
